package com.atguigu.gmall.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.app.func.BeanToJsonStrFunction;
import com.atguigu.gmall.realtime.app.func.DimAsyncFunction;
import com.atguigu.gmall.realtime.beans.TradeSkuOrderBean;
import com.atguigu.gmall.realtime.utils.DateFormatUtil;
import com.atguigu.gmall.realtime.utils.DorisUtil;
import com.atguigu.gmall.realtime.utils.KafkaUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.math.BigDecimal;
import java.util.concurrent.TimeUnit;

/**
 * @author 黄凯
 * @date 2023/7/17
 * sku粒度下单业务过程聚合统计
 * 需要启动的进程
 *      zk、kafka、maxwell、hdfs、hbase、redis、doris、DwdTradeOrderDetail、DwsTradeSkuOrderWindow
 * 开发流程
 *      准备环境
 *      检查点相关的设置
 *      从kafka的下单事实表中读取数据
 *      过滤掉空消息 对读取的数据进行类型转换 jsonStr->jsonObj
 *      去重
 *          为什么会产生空消息以及重复数据？
 *              因为我们是从dwd的下单事实表中读取数据，下单事实表是将订单明细表、订单表、订单明细活动表、
 *              订单明细优惠券表这四张表进行关联，在关联明细活动以及明细优惠券的时候，使用了左外连接。
 *              对于左外连接，如果左表数据先到，右表数据后到的情况，会产生如下三条数据
 *              [左表     null]   +I
 *              [左表     null]   -D
 *              [左表     右表]    +I
 *              这样的数据，发送到kafka主题后，kafka主题会接收到如下三条消息
 *              [左表     null]
 *              null
 *              [左表     右表]
 *          去重实现
 *              按照唯一键(订单明细id)进行分组
 *              方式1：状态 + 定时器
 *                  当第一条数据来的时候，将其放到状态中，并注册5s后执行的定时器
 *                  当第二条数据来的时候，我们会用这条数据和状态中的数据进行join时间比较，将时间大的放到状态中
 *                  当定时器被触发执行的时候，将状态中的内容传递到下游
 *                  优点：如果数据重复了，传输的数据只有一条
 *                  缺点：时效性差
 *              方式2：状态 + 抵消
 *                  当第一条数据来的时候，直接将其发送到下游，并放到状态中
 *                  当第二条数据来的时候，将状态中影响到度量值的字段进行取反，发送到下游
 *                  将第二条数据也发送到下游去
 *                  优点：时效性好
 *                  缺点：如果数据重复了，传输的数据有三条
 *      再次对流中数据进行类型转换   jsonObj->实体类对象(起始结束时间、维度、度量)
 *      指定Watermark以及提取事件时间字段
 *      分组
 *      开窗
 *      聚合计算
 *      维度关联
 *          基本的维度关联的实现
 *              HbaseUtil->JSONObject getDimInfoFromHbase
 *          优化1：旁路缓存
 *              思路：先从缓存中查询维度数据，如果在缓存中找到了要关联的维度，直接将其作为方法的返回值进行返回(缓存命中)
 *                   如果在缓存中没有找到对应的维度信息，发送请求到hbase进行查询，并将查询的结果放到缓存中，方便下次查询
 *              缓存产品选型
 *                  状态： 性能好、维护性差
 *                  redis：性能也不错、维护性好    √
 *              关于Redis的设置：
 *                  key:    维度表名:主键值
 *                  type:   string
 *                  expire: 1day    避免冷数据常驻内存，给内存带来压力
 *                  注意：如果维度数据发生了变化，将Redis中缓存的数据删除掉
 *                  DimUtil -> JSONObject getDimInfo
 *          优化2：异步IO
 *  *              在对流中数据进行处理数据的时候，如果使用的map算子，底层是同步的处理方式，要想提升处理能力，
 *  *              可以将并行度提升一下，但是更大的并行度意味需要更多的资源，不可能无限制的提升。
 *  *              Flink提供的异步处理的API，可以对流中的数据进行异步的处理
 *  *              AsyncDataStream.[un]orderedWait(
 *  *                  流，
 *  *                  如何发送异步请求， implements AsyncFunction
 *  *                  超时时间,
 *  *                  时间单位
 *  *              )
 *              interface:DimJoinFunction
 *                  getKey-根据流中的对象获取维度主键
 *                  join  -将维度对象的属性补充到流中对象上
 *              class DimAsyncFunction extends RichAsyncFunction implements DimJoinFunction{
 *                  open:初始化连接
 *                      Redis异步连接
 *                      Hbase异步连接
 *                  close:关闭链接
 *                  asyncInvoke:
 *                      CompletableFuture
 *                          .supplyAsync  执行CompletableFuture任务，支持返回值
 *                              发送请求到Redis中获取数据
 *                          第一个任务执行完成后，执行第二个回调方法任务，会将该任务的执行结果，
 *                          作为入参，传递到回调方法中，并且回调方法是有返回值的
 *                          .thenApplyAsync
 *                              根据Redis中查询的结果，到Hbase获取数据
 *                          .thenAcceptAsync
 *                              将维度对象的属性补充到流中的对象上
 *                              将补充完维度的对象传递到下游
 *              }
 *      将结果写到Doris中
 */
public class DwsTradeSkuOrderWindow {
    public static void main(String[] args) throws Exception {
        //TODO 1.基本环境准备
        //1.1 指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(4);
        //TODO 2.检查点相关设置
        env.enableCheckpointing(5000L);

        //TODO 3.从kafka的主题中读取数据
        //3.1 声明消费的主题以及消费者组
        String topic = "dwd_trade_order_detail";
        String groupId = "dws_trade_sku_order_group";
        //3.2 创建消费者对象
        KafkaSource<String> kafkaSource = KafkaUtil.getKafkaSource(topic, groupId);
        //3.3 消费数据 封装为流
        DataStreamSource<String> kafkaStrDS
                = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafka_source");

        //TODO 4.过滤空消息  并对流中的数据类型进行转换   jsonStr->jsonObj
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaStrDS.process(
                new ProcessFunction<String, JSONObject>() {
                    @Override
                    public void processElement(String jsonStr, Context ctx, Collector<JSONObject> out) throws Exception {
                        if (StringUtils.isNotEmpty(jsonStr)) {
                            out.collect(JSON.parseObject(jsonStr));
                        }
                    }
                }
        );
        // {"create_time":"2023-07-17 09:30:40","sku_num":"1","split_original_amount":"129.0000",
        // "split_coupon_amount":"19.55","sku_id":"27","date_id":"2023-07-17","coupon_id":"1",
        // "user_id":"3119","province_id":"31","sku_name":"索唇膏 ","id":"14256140","order_id":"53814",
        // "split_activity_amount":"0.0","split_total_amount":"109.45","ts":"1689557440"}
        // jsonObjDS.print(">>>");

        //TODO 5.去重
        //5.1 按照唯一键进行分组(订单明细的id)
        KeyedStream<JSONObject, String> keyedOrderDetailIdDS
                = jsonObjDS.keyBy(jsonObj -> jsonObj.getString("id"));
        //5.2 去重方式1： 状态编程 + 定时器  不足：每条数据要等5s后向下游传递，影响时效性
        /*keyedOrderDetailIdDS.process(
            new KeyedProcessFunction<String, JSONObject, JSONObject>() {
                private ValueState<JSONObject> lastJsonObjState;

                @Override
                public void open(Configuration parameters) throws Exception {
                    ValueStateDescriptor<JSONObject> valueStateDescriptor
                        = new ValueStateDescriptor<JSONObject>("valueStateDescriptor", JSONObject.class);
                    lastJsonObjState = getRuntimeContext().getState(valueStateDescriptor);
                }

                @Override
                public void processElement(JSONObject jsonObj, Context ctx, Collector<JSONObject> out) throws Exception {
                    //从状态中获取数据
                    JSONObject lastJsonObj = lastJsonObjState.value();
                    if (lastJsonObj == null) {
                        //说明目前没有产生重复，将当前数据放到状态中
                        lastJsonObjState.update(jsonObj);
                        //注册一个5秒钟之后执行的定时器
                        long currentProcessingTime = ctx.timerService().currentProcessingTime();
                        ctx.timerService().registerProcessingTimeTimer(currentProcessingTime + 5000L);
                    } else {
                        //说明产生重复了，用当前这条数据的聚合时间和状态中存在的数据的聚合时间进行比较，将时间大的放到状态中
                        //伪代码  因为我们在DWD层我们没有传递聚合时间过来
                        String ts1 = lastJsonObj.getString("ts");
                        String ts2 = jsonObj.getString("ts");
                        if(ts2.compareTo(ts1) >=0){
                            lastJsonObjState.update(jsonObj);
                        }
                    }
                }

                @Override
                public void onTimer(long timestamp, OnTimerContext ctx, Collector<JSONObject> out) throws Exception {
                    //定时器被触发  会执行当前方法
                    JSONObject jsonObj = lastJsonObjState.value();
                    out.collect(jsonObj);
                    //清除状态
                    lastJsonObjState.clear();
                }
            }
        );*/
        //5.3 去重方式2: 状态编程 + 抵消
        SingleOutputStreamOperator<JSONObject> distinctDS = keyedOrderDetailIdDS.process(
                new KeyedProcessFunction<String, JSONObject, JSONObject>() {
                    private ValueState<JSONObject> lastJsonObjState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ValueStateDescriptor<JSONObject> valueStateDescriptor
                                = new ValueStateDescriptor<JSONObject>("valueStateDescriptor", JSONObject.class);
                        valueStateDescriptor.enableTimeToLive(StateTtlConfig.newBuilder(Time.seconds(10)).build());
                        lastJsonObjState = getRuntimeContext().getState(valueStateDescriptor);
                    }

                    @Override
                    public void processElement(JSONObject jsonObj, Context ctx, Collector<JSONObject> out) throws Exception {
                        JSONObject lastJsonObj = lastJsonObjState.value();
                        if (lastJsonObj != null) {
                            //说明是有重复数据  将状态中影响到度量值的字段进行取反发送到下游
                            String splitOriginalAmount = lastJsonObj.getString("split_original_amount");
                            String splitCouponAmount = lastJsonObj.getString("split_coupon_amount");
                            String splitActivityAmount = lastJsonObj.getString("split_activity_amount");
                            String splitTotalAmount = lastJsonObj.getString("split_total_amount");
                            lastJsonObj.put("split_original_amount", "-" + splitOriginalAmount);
                            lastJsonObj.put("split_coupon_amount", "-" + splitCouponAmount);
                            lastJsonObj.put("split_activity_amount", "-" + splitActivityAmount);
                            lastJsonObj.put("split_total_amount", "-" + splitTotalAmount);
                            out.collect(lastJsonObj);
                        }
                        out.collect(jsonObj);
                        lastJsonObjState.update(jsonObj);
                    }
                }
        );

        // distinctDS.print(">>>>");

        //TODO 6.再次类型转换     jsonObj->实体类对象
        SingleOutputStreamOperator<TradeSkuOrderBean> orderBeanDS = distinctDS.map(
                new MapFunction<JSONObject, TradeSkuOrderBean>() {
                    @Override
                    public TradeSkuOrderBean map(JSONObject jsonObj) throws Exception {
                        // {"create_time":"2023-07-17 09:30:40","sku_num":"1","split_original_amount":"129.0000",
                        // "split_coupon_amount":"19.55","sku_id":"27","date_id":"2023-07-17","coupon_id":"1",
                        // "user_id":"3119","province_id":"31","sku_name":"索唇膏 ","id":"14256140","order_id":"53814",
                        // "split_activity_amount":"0.0","split_total_amount":"109.45","ts":"1689557440"}
                        String skuId = jsonObj.getString("sku_id");
                        String splitOriginalAmount = jsonObj.getString("split_original_amount");
                        String splitCouponAmount = jsonObj.getString("split_coupon_amount");
                        String splitActivityAmount = jsonObj.getString("split_activity_amount");
                        String splitTotalAmount = jsonObj.getString("split_total_amount");
                        Long ts = jsonObj.getLong("ts") * 1000;
                        TradeSkuOrderBean orderBean = TradeSkuOrderBean.builder()
                                .skuId(skuId)
                                .originalAmount(new BigDecimal(splitOriginalAmount))
                                .couponAmount(new BigDecimal(splitCouponAmount))
                                .activityAmount(new BigDecimal(splitActivityAmount))
                                .orderAmount(new BigDecimal(splitTotalAmount))
                                .ts(ts)
                                .build();
                        return orderBean;
                    }
                }
        );

        // orderBeanDS.print(">>>");

        //TODO 7.指定Watermark以及提取事件时间字段
        SingleOutputStreamOperator<TradeSkuOrderBean> withWatermarkDS = orderBeanDS.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .<TradeSkuOrderBean>forMonotonousTimestamps()
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<TradeSkuOrderBean>() {
                                    @Override
                                    public long extractTimestamp(TradeSkuOrderBean orderBean, long recordTimestamp) {
                                        return orderBean.getTs();
                                    }
                                }
                        )
        );
        //TODO 8.按照维度进行分组(sku_id)
        KeyedStream<TradeSkuOrderBean, String> keyedSkuIdDS
                = withWatermarkDS.keyBy(TradeSkuOrderBean::getSkuId);

        //TODO 9.开窗
        WindowedStream<TradeSkuOrderBean, String, TimeWindow> windowDS
                = keyedSkuIdDS.window(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10)));

        //TODO 10.聚合计算
        SingleOutputStreamOperator<TradeSkuOrderBean> reduceDS = windowDS.reduce(
                new ReduceFunction<TradeSkuOrderBean>() {
                    @Override
                    public TradeSkuOrderBean reduce(TradeSkuOrderBean value1, TradeSkuOrderBean value2) throws Exception {
                        value1.setOriginalAmount(value1.getOriginalAmount().add(value2.getOriginalAmount()));
                        value1.setActivityAmount(value1.getActivityAmount().add(value2.getActivityAmount()));
                        value1.setCouponAmount(value1.getCouponAmount().add(value2.getCouponAmount()));
                        value1.setOrderAmount(value1.getOrderAmount().add(value2.getOrderAmount()));
                        return value1;

                    }
                },
                new WindowFunction<TradeSkuOrderBean, TradeSkuOrderBean, String, TimeWindow>() {
                    @Override
                    public void apply(String s, TimeWindow window, Iterable<TradeSkuOrderBean> input, Collector<TradeSkuOrderBean> out) throws Exception {
                        String stt = DateFormatUtil.toYmdHms(window.getStart());
                        String edt = DateFormatUtil.toYmdHms(window.getEnd());
                        String curDate = DateFormatUtil.toPartitionDate(window.getStart());
                        for (TradeSkuOrderBean orderBean : input) {
                            orderBean.setStt(stt);
                            orderBean.setEdt(edt);
                            orderBean.setCurDate(curDate);
                            out.collect(orderBean);
                        }
                    }
                }
        );
        // reduceDS.print(">>>>");

        //TODO 11.关联sku维度
        /*SingleOutputStreamOperator<TradeSkuOrderBean> withSkuInfoDS = reduceDS.map(
            new DimMapFunction<TradeSkuOrderBean>("dim_sku_info") {
                @Override
                public void join(TradeSkuOrderBean orderBean, JSONObject dimJsonObj) {
                    orderBean.setSkuName(dimJsonObj.getString("sku_name"));
                    orderBean.setSpuId(dimJsonObj.getString("spu_id"));
                    orderBean.setTrademarkId(dimJsonObj.getString("tm_id"));
                    orderBean.setCategory3Id(dimJsonObj.getString("category3_id"));
                }

                @Override
                public String getKey(TradeSkuOrderBean orderBean) {
                    return orderBean.getSkuId();
                }
            }
        );
        // withSkuInfoDS.print(">>>>");*/

        //将异步I/O操作应用于DataStream作为DataStream的一次转换操作
        SingleOutputStreamOperator<TradeSkuOrderBean> withSkuInfoDS = AsyncDataStream.unorderedWait(
                reduceDS,
                //发送异步请求的过程，实现分发请求的AsyncFunction
                new DimAsyncFunction<TradeSkuOrderBean>("dim_sku_info") {
                    @Override
                    public void join(TradeSkuOrderBean orderBean, JSONObject dimJsonObj) {
                        orderBean.setSkuName(dimJsonObj.getString("sku_name"));
                        orderBean.setSpuId(dimJsonObj.getString("spu_id"));
                        orderBean.setTrademarkId(dimJsonObj.getString("tm_id"));
                        orderBean.setCategory3Id(dimJsonObj.getString("category3_id"));
                    }

                    @Override
                    public String getKey(TradeSkuOrderBean orderBean) {
                        return orderBean.getSkuId();
                    }
                },
                60,
                TimeUnit.SECONDS
        );
        // withSkuInfoDS.print(">>>>>");

        //TODO 12.关联spu维度
        SingleOutputStreamOperator<TradeSkuOrderBean> withSpuInfoDS = AsyncDataStream.unorderedWait(
                withSkuInfoDS,
                new DimAsyncFunction<TradeSkuOrderBean>("dim_spu_info") {
                    @Override
                    public void join(TradeSkuOrderBean orderBean, JSONObject dimInfo) {
                        orderBean.setSpuName(dimInfo.getString("spu_name"));
                    }

                    @Override
                    public String getKey(TradeSkuOrderBean orderBean) {
                        return orderBean.getSpuId();
                    }
                },
                60,
                TimeUnit.SECONDS
        );
        //TODO 13.关联tm维度
        SingleOutputStreamOperator<TradeSkuOrderBean> withTmDS = AsyncDataStream.unorderedWait(
                withSpuInfoDS,
                new DimAsyncFunction<TradeSkuOrderBean>("dim_base_trademark") {
                    @Override
                    public void join(TradeSkuOrderBean orderBean, JSONObject dimInfo) {
                        orderBean.setTrademarkName(dimInfo.getString("tm_name"));
                    }

                    @Override
                    public String getKey(TradeSkuOrderBean orderBean) {
                        return orderBean.getTrademarkId();
                    }
                },
                60,
                TimeUnit.SECONDS
        );
        //TODO 14.关联category3维度
        SingleOutputStreamOperator<TradeSkuOrderBean> c3Stream = AsyncDataStream.unorderedWait(
                withTmDS,
                new DimAsyncFunction<TradeSkuOrderBean>("dim_base_category3") {

                    @Override
                    public String getKey(TradeSkuOrderBean bean) {
                        return bean.getCategory3Id();
                    }

                    @Override
                    public void join(TradeSkuOrderBean bean,
                                     JSONObject dim) {
                        bean.setCategory3Name(dim.getString("name"));
                        bean.setCategory2Id(dim.getString("category2_id"));
                    }
                },
                60,
                TimeUnit.SECONDS
        );


        //TODO 15.关联category2维度
        SingleOutputStreamOperator<TradeSkuOrderBean> c2Stream = AsyncDataStream.unorderedWait(
                c3Stream,
                new DimAsyncFunction<TradeSkuOrderBean>("dim_base_category2") {

                    @Override
                    public String getKey(TradeSkuOrderBean bean) {
                        return bean.getCategory2Id();
                    }

                    @Override
                    public void join(TradeSkuOrderBean bean,
                                     JSONObject dim) {
                        bean.setCategory2Name(dim.getString("name"));
                        bean.setCategory1Id(dim.getString("category1_id"));
                    }
                },
                60,
                TimeUnit.SECONDS
        );

        //TODO 16.关联category1维度
        SingleOutputStreamOperator<TradeSkuOrderBean> restulStream = AsyncDataStream.unorderedWait(
                c2Stream,
                new DimAsyncFunction<TradeSkuOrderBean>("dim_base_category1") {

                    @Override
                    public String getKey(TradeSkuOrderBean bean) {
                        return bean.getCategory1Id();
                    }

                    @Override
                    public void join(TradeSkuOrderBean bean,
                                     JSONObject dim) {
                        bean.setCategory1Name(dim.getString("name"));
                    }
                },
                60,
                TimeUnit.SECONDS
        );

        //TODO 17.将关联的结果写到Doris中
        restulStream.print(">>>>");
        restulStream
                .map(new BeanToJsonStrFunction<>())
                .sinkTo(DorisUtil.getDorisSink("dws_trade_sku_order_window"));

        env.execute();
    }
}
